package com.bjsxt.receive;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import com.rabbitmq.client.Channel;
import java.io.IOException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

/**
 * Created with IntelliJ IDEA.
 *
 * @Auther: 雷哥
 * @Date: 2020/10/11/14:49
 * @Description:
 */
@Component
public class MessageReceive4 {

  @Autowired
  private StringRedisTemplate redisTemplate;

  @Autowired
  private BitMapBloomFilter bitMapBloomFilter;

  //消息的前缀
  private String MESSAGE="message:";

  @RabbitListener(bindings = {
      @QueueBinding(value = @Queue("queue"),
      key = "error",
      exchange = @Exchange(value = "directs"))//默认的直连交换机
  })
  public void receiveMessage(String content, Message message, Channel channel) throws IOException {
    long deliveryTag = message.getMessageProperties().getDeliveryTag();//消息投递ID
    System.out.println("消息投递ID:"+deliveryTag);
    String messageId = message.getMessageProperties().getMessageId();
    System.out.println("消息自定义ID:"+messageId);

    if(bitMapBloomFilter.contains(messageId)){
      System.out.println("这个消息被消费过，不能重复消费");
      try {
        //如果进入到这里面，说明这个消息之前被消费过，但是MQ认为你没有消费，所以我们要签收这条消息
        channel.basicAck(deliveryTag,false);
        return;
      }catch (Exception e){
        System.out.println(e);
      }
    }

    if(content.equals("123456")){//如果消息内容为123456就签收它
      /**
       * 参数说明:
       * deliveryTag  消息投递ID，要签收的投递ID是多少
       * multiple:是否批量签收
       */
      channel.basicAck(deliveryTag,false);
      System.out.println("消息签收成功");
      //消费成功之后放到布隆过滤器里面
      bitMapBloomFilter.add(messageId);
    }else{
      String count=redisTemplate.opsForValue().get(MESSAGE+messageId);
      if(count!=null&&Long.valueOf(count)>=3){
        channel.basicNack(deliveryTag,false,false);
        System.out.println("该消息消费3失败，我们记录它，人工处理:"+content);
      }else {
        //如果不是123456就拒绝签收
        /**
         * 参数说明:
         * deliveryTag  消息投递ID，要签收的投递ID是多少
         * multiple:是否批量签收
         * requeue: true 代表拒绝签收并把消息重新放回到队列里面  false就直接拒绝
         */
        //处理业务逻辑的处理[可能逻辑出现问题]
        channel.basicNack(deliveryTag, false, true);
        System.out.println("消息被拒绝签收");
        //因现被拒绝了，我们把消息ID放到redis里面
        redisTemplate.opsForValue().increment(MESSAGE + messageId);
      }
    }
    System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~");
  }

}
